package com.scut.campus.infra.mq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Collections;
import java.util.List;

@Slf4j
@Component
public class RocketMQProducer {

    private static DefaultMQProducer producer;

    @Value("${spring.rocketmq.name-server}")
    private String namesrvAddr;

    @Value("${spring.rocketmq.consumer.group_comment}")
    private String producerGroup;

    public RocketMQProducer(){

    }

    @PostConstruct
    public void start() throws MQClientException {
        producer = new DefaultMQProducer();
        producer.setNamesrvAddr(namesrvAddr);
        producer.setProducerGroup(producerGroup);
        producer.start();
        log.info("RocketMQ Producer started:{}", namesrvAddr);
    }

    public <T> SendResult sendMessage(String topic, T messageBody) {

        try {
            // 1. FastJSON序列化（启用类型信息）
            byte[] jsonBytes = JSON.toJSONBytes(messageBody,
                    SerializerFeature.WriteClassName,
                    SerializerFeature.DisableCircularReferenceDetect
            );

            // 2. 构建消息对象
            Message message = new Message(topic, jsonBytes);

            // 3. 发送消息
            SendResult sendResult = producer.send(message);

            log.info(" 消息发送成功 | msgId={} | body={}",
                    sendResult.getMsgId(),  JSON.toJSONString(messageBody));
            return sendResult;

        } catch (com.alibaba.fastjson.JSONException  e) {
            log.error("FastJSON 序列化失败 | topic={} | type={}",
                    topic, messageBody.getClass().getName(),  e);
            throw new IllegalArgumentException("消息体序列化异常", e);
        } catch (MQClientException e) {
            log.error("RocketMQ 客户端异常 | topic={}", topic, e);
            throw new RuntimeException("MQ服务不可用", e);
        } catch (RemotingException | InterruptedException e) {
            log.error(" 消息发送中断 | topic={}", topic, e);
            Thread.currentThread().interrupt();
            throw new RuntimeException("网络通信异常", e);
        } catch (MQBrokerException e) {
            throw new RuntimeException(e);
        }


    }

    @PreDestroy
    public void shutdown() {
        producer.shutdown();
        System.out.println("Producer shut down.");
    }
}
